package rocketmq

import (
	"context"
	"errors"
	"fmt"
	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/apache/rocketmq-client-go/v2/producer"
)

type MqProducer struct {
	topicName string
	producer  rocketmq.Producer
	RetryNum  int
}

// NewMQProducer 生产者
func NewMQProducer(topicName, groupName string, NameSrvAddr []string) *MqProducer {
	p, err := rocketmq.NewProducer(
		//服务地址
		producer.WithNameServer(NameSrvAddr),
		//组名
		producer.WithGroupName(groupName),
		//重试次数
		producer.WithRetry(1),
	)
	if err != nil {
		panic(err)
	}
	err = p.Start()
	if err != nil {
		panic(err)
	}
	return &MqProducer{
		topicName: topicName,
		producer:  p,
		RetryNum:  5,
	}
}

// Shutdown 关闭
func (mq *MqProducer) Shutdown(err error) error {
	if err != nil {
		if mq.RetryNum >= 0 {
			fmt.Printf("发送消息失败: %v \n", err.Error())
			fmt.Printf("重试次数: %v \n", mq.RetryNum)
			mq.RetryNum -= 1
			if mq.RetryNum <= 1 {
				err = mq.producer.Shutdown()
				if err != nil {
					fmt.Println("Close MQProducer error", err)
				}
			}
		}
	} else {
		mq.RetryNum = 5
	}
	return err
}

// SendSyncMessage 发送同步消息
func (mq *MqProducer) SendSyncMessage(msg *MQMessage) (bool, error) {
	if mq.RetryNum < 0 {
		return false, errors.New("已关闭MQ")
	}
	msg.TopicName = mq.topicName
	res, err := mq.producer.SendSync(context.Background(), msg.toMessage())
	err = mq.Shutdown(err)
	if err != nil {
		return false, err
	}
	fmt.Printf("send sync message success. result=%s\n", res.String())
	return true, nil
}

// SendSyncBatchMessage 发送批量消息
func (mq *MqProducer) SendSyncBatchMessage(msgs []*MQMessage) (bool, error) {
	if mq.RetryNum < 0 {
		return false, errors.New("已关闭MQ")
	}
	mqmgs := make([]*primitive.Message, 0)
	for _, msg := range msgs {
		msg.TopicName = mq.topicName
		mqmgs = append(mqmgs, msg.toMessage())
	}
	res, err := mq.producer.SendSync(context.Background(), mqmgs...)
	err = mq.Shutdown(err)
	if err != nil {
		return false, err
	}
	fmt.Printf("send sync message success. result=%s\n", res.String())
	return true, nil
}

// SendAsyncMessageBody 发送异步消息后回调
func (mq *MqProducer) SendAsyncMessageBody(msg *MQMessage) (bool, error) {
	if mq.RetryNum < 0 {
		return false, errors.New("已关闭MQ")
	}
	err := mq.producer.SendAsync(
		context.Background(),
		func(ctx context.Context, result *primitive.SendResult, e error) {
			if e != nil {
				fmt.Printf("receive message send error: %s\n", e)
				return
			}
		}, primitive.NewMessage(mq.topicName, msg.Body))

	err = mq.Shutdown(err)
	if err != nil {
		return false, err
	}
	return true, nil
}

// SendAsyncMessage 发送异步消息后回调
func (mq *MqProducer) SendAsyncMessage(msg *MQMessage,
	callback func(ctx context.Context, result *primitive.SendResult, err error)) (bool, error) {
	if mq.RetryNum < 1 {
		return false, fmt.Errorf("已关闭MQ")
	}
	f := func(ctx context.Context, result *primitive.SendResult, err error) {
		if err != nil {
			er := fmt.Errorf("异步发送消息失败: %v", err)
			err = mq.Shutdown(er)
		}
		callback(ctx, result, err)
		return
	}
	msg.TopicName = mq.topicName
	err := mq.producer.SendAsync(context.Background(), f, msg.toMessage())
	err = mq.Shutdown(err)
	if err != nil {
		return false, err
	}
	return true, nil
}
